package com.sardine.base.hbase.executor;

import com.sardine.base.hbase.persister.HEntity;
import com.sardine.base.hbase.persister.HEntityAdhesive;
import com.sardine.base.hbase.persister.HEntitySkeleton;
import com.sardine.base.hbase.persister.HEntitySkeletonResolver;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.List;

/**
 * 上海京东到家元信信息技术有限公司<br>
 * Created with IntelliJ IDEA.
 * Date:2017/11/21
 * Time:13:33
 *
 * @author daifeng<br>
 *         daifeng@imdada.cn
 * @since JDK 1.7 <br>
 * Copyright (c) 2017 imdada System Incorporated All Rights Reserved.
 */
public class DefaultHBaseExecutor implements HBaseExecutor {
    private static final Logger logger = LoggerFactory.getLogger(DefaultHBaseExecutor.class);
    private HEntitySkeletonResolver skeletonResolver = new HEntitySkeletonResolver();
    private HBaseRepository hBaseRepository;
    private HEntityAdhesive HEntityAdhesive;

    public DefaultHBaseExecutor(HEntityAdhesive HEntityAdhesive, HBaseRepository hBaseRepository) {
        this.HEntityAdhesive = HEntityAdhesive;
        this.hBaseRepository = hBaseRepository;
    }

    /**
     * 插入一条记录
     *
     * @param o
     * @return
     */
    @Override
    public boolean put(Object o) {
        HEntitySkeleton skeleton = skeletonResolver.getSkeleton(o.getClass());
        HEntity entity = HEntityAdhesive.getPerSister(skeleton, o);
        hBaseRepository.save(entity);
        return true;
    }

    /**
     * 同步批量插入
     *
     * @param objList
     * @return
     */
    @Override
    public boolean put(List<? extends Object> objList) {
        List<HEntity> hEntities = new ArrayList<>();
        for (Object o : objList) {
            HEntitySkeleton skeleton = skeletonResolver.getSkeleton(o.getClass());
            HEntity entity = HEntityAdhesive.getPerSister(skeleton, o);
            hEntities.add(entity);
        }
        hBaseRepository.save(hEntities);
        return true;
    }

    /**
     * 异步批量插入
     *
     * @param objList
     * @return
     */
    @Override
    public boolean putAsync(List<? extends Object> objList) {
        List<HEntity> hEntities = new ArrayList<>();
        for (Object o : objList) {
            HEntitySkeleton skeleton = skeletonResolver.getSkeleton(o.getClass());
            HEntity entity = HEntityAdhesive.getPerSister(skeleton, o);
            hEntities.add(entity);
        }
        hBaseRepository.saveAsync(hEntities);
        return true;
    }

    /**
     * 查询一条记录rowKey
     *
     * @param rowKey
     * @param aClass
     * @return
     */
    @Override
    public <T> T getOne(Serializable rowKey, Class<T> aClass) {
        HEntitySkeleton<T> skeleton = skeletonResolver.getSkeleton(aClass);
        HEntity hEntity = HEntityAdhesive.getPerRowKey(rowKey, skeleton);
        hEntity = hBaseRepository.load(hEntity);
        return HEntityAdhesive.decoderModel(skeleton, hEntity);
    }

    /**
     * 批量查询一组rowkey
     *
     * @param rowKeyList
     * @param aClass
     * @return
     */
    @Override
    public <T> List<T> getList(List<? extends Serializable> rowKeyList, Class<T> aClass) {
        HEntitySkeleton<T> skeleton = skeletonResolver.getSkeleton(aClass);
        HEntity hEntity = null;
        List<String> rowKeys = new ArrayList<>();
        for (Serializable serializable : rowKeyList) {
            hEntity = HEntityAdhesive.getPerRowKey(serializable, skeleton);
            rowKeys.add(hEntity.getRowKey());
        }
        List<HEntity> hEntities = hBaseRepository.loadList(hEntity, rowKeys);
        List<T> list = new ArrayList<>();
        if (hEntities != null) {
            for (HEntity entity : hEntities) {
                T t = HEntityAdhesive.decoderModel(skeleton, entity);
                list.add(t);
            }
        }
        return list;
    }
}
